package studio.raptor.extractor.segment.server;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.extractor.common.constant.EventMetaStatus;
import studio.raptor.extractor.common.holder.ExecutorHolder;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventDataInfo;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;
import studio.raptor.extractor.segment.task.BuildJsonTask;
import studio.raptor.extractor.segment.task.EventActiveTask;
import studio.raptor.extractor.segment.task.EventDeliveringTask;
import studio.raptor.extractor.segment.task.EventMQExtractingTask;
import studio.raptor.extractor.segment.task.EventRecoverTask;
import studio.raptor.extractor.segment.task.NodeMsgAssemblingTask;

/**
 * Created by liuyong8 on 2017/7/12.
 */
@Component
@Scope("prototype")
public class ExtractorServer {

  private static final Logger log = LoggerFactory.getLogger(ExtractorServer.class);

  @Autowired
  ApplicationContext context;

  // 存放带抽文的eventmeta队列
  private BlockingQueue<EventMeta> eventQueue;
  // 存放节点报文
  private BlockingQueue<EventDataInfo> evenDataQueue;
  // 组装队列
  private BlockingQueue<BuildJsonResult> buildJsonQueue;

  private ServerConfig serverConfig;

  public static final int coreSize = Runtime.getRuntime().availableProcessors();

  public static final int maxSize = Runtime.getRuntime().availableProcessors() * 40;

  //检查剩余未处理event
  private ExecutorService checkRecoverProgress;
  //事件报文生产者线程池
  private ExecutorService extractingproducerExecutor;
  // 事件报文抽取线程池
  private ExecutorService extractingExecutor;
  //事件激活线程池
  private ExecutorService activeExecutor;
  //节点报文线程池
  private ExecutorService assemblingExecutor;
  //event报文组装线程池
  private ExecutorService buildingExecutor;
  //节点报文组装线程池
  private ExecutorService nodeBuildingExecutor;
  //发送线程池
  private ExecutorService deliveringExecutor;

  @Autowired
  private EventMetaService eventMetaService;

  private void init() {
    eventQueue = new LinkedBlockingQueue<EventMeta>(serverConfig.getEventQueueSize());
    evenDataQueue = new LinkedBlockingQueue<EventDataInfo>(serverConfig.getEventDataQueueSize());
    buildJsonQueue = new LinkedBlockingQueue<>(serverConfig.getBuildJsonQueueSize());

//    checkRecoverProgress = ExecutorHolder.createSingleThreadPool("check_recover_progress" + "_" + serverConfig.getKindId());
    extractingproducerExecutor = ExecutorHolder
        .createSingleThreadPool("event_producter_extracting" + "_" + serverConfig.getKindId());
    extractingExecutor = ExecutorHolder
        .createSingleThreadPool("event_mq_extracting" + "_" + serverConfig.getKindId());
    activeExecutor = ExecutorHolder
        .createSingleThreadPool("event_activing" + "_" + serverConfig.getKindId());
    assemblingExecutor = ExecutorHolder
        .createCachedThreadPool("node_msg_assembling" + "_" + serverConfig.getKindId(), coreSize,
            maxSize);
    buildingExecutor = ExecutorHolder
        .createSingleThreadPool("event_building" + "_" + serverConfig.getKindId());
    nodeBuildingExecutor = ExecutorHolder
        .createCachedThreadPool("nod_building" + "_" + serverConfig.getKindId(), coreSize, maxSize);
    deliveringExecutor = ExecutorHolder
        .createSingleThreadPool("msg_delivering" + "_" + serverConfig.getKindId());

  }

  /**
   * 启动
   * @throws Exception 异常
   */
  public void startup() throws Exception {
    init();
//    recover();

//  extractingExecutor.submit(new EventRecoverTask());
//  extractingproducerExecutor.submit(new EventMQProduterTask());
    extractingExecutor.submit(createMQExtractingTask());
//  activeExecutor.submit(createEventActiveTask());
    assemblingExecutor.submit(createNodeMsgAssemblingTask());
    buildingExecutor.submit(createBuilJsonTask());
    deliveringExecutor.submit(createEventDeliveringTask());

  }

  /**
   * 关闭
   */
  public void shutDown(){

  }

  private EventRecoverTask createEventRecoverTask(){
    EventRecoverTask eventRecoverTask = context.getBean(EventRecoverTask.class);
    eventRecoverTask.setServerConfig(serverConfig);
    return eventRecoverTask;
  }

  private EventDeliveringTask createEventDeliveringTask() {
    EventDeliveringTask eventDeliveringTask = context.getBean(EventDeliveringTask.class);
    eventDeliveringTask.setBuildJsonQueue(buildJsonQueue);
    eventDeliveringTask.setServerConfig(serverConfig);
    return eventDeliveringTask;
  }

  private BuildJsonTask createBuilJsonTask() {
    BuildJsonTask buildJsonTask = context.getBean(BuildJsonTask.class);
    buildJsonTask.setBuildJsonQueue(buildJsonQueue);
    buildJsonTask.setEventDataQueue(evenDataQueue);
    buildJsonTask.setNodeBuildingExecutor(nodeBuildingExecutor);
    buildJsonTask.setServerConfig(serverConfig);
    return buildJsonTask;
  }

  private NodeMsgAssemblingTask createNodeMsgAssemblingTask() {
    NodeMsgAssemblingTask nodeMsgAssemblingTask = context.getBean(NodeMsgAssemblingTask.class);
    nodeMsgAssemblingTask.setEventMetaQueue(eventQueue);
    nodeMsgAssemblingTask.setEventDataQueue(evenDataQueue);
    nodeMsgAssemblingTask.setExec(assemblingExecutor);
    nodeMsgAssemblingTask.setServerConfig(serverConfig);
    return nodeMsgAssemblingTask;
  }

  private EventMQExtractingTask createMQExtractingTask() {
    EventMQExtractingTask eventMQExtractingTask = context.getBean(EventMQExtractingTask.class);
    eventMQExtractingTask.setServerConfig(this.getServerConfig());
    eventMQExtractingTask.setEventQueue(eventQueue);
    return eventMQExtractingTask;
  }

  private EventActiveTask createEventActiveTask() {
    EventActiveTask eventActiveTask = context.getBean(EventActiveTask.class);
    eventActiveTask.setServerConfig(this.getServerConfig());
    eventActiveTask.setEventQueue(this.getEventQueue());
    return eventActiveTask;
  }

  /**
   * recover 入口
   */
  private void recover() throws Exception {
    serverConfig.setExtractorMQEnabled(new AtomicBoolean(false));
    // step1:recover event meta P、F -> W
    List<String> statuses = new ArrayList<>();
    statuses.add(EventMetaStatus.STATUS_PROCESS.value());
    statuses.add(EventMetaStatus.STATUS_FALSE.value());
    statuses.add(EventMetaStatus.STATUS_WAIT.value());
    List<EventMeta> eventMetas = eventMetaService
        .findMetasByKindIdAndStatuses(serverConfig.getKindId(), statuses);
    if (eventMetas.isEmpty()) {
      log.info("events in processing is empty，nothing to recover!");
    } else {
      log.info("begin try to reset ordered-event state:[P、F->W]", eventMetas.size());
      for (EventMeta eventMeta : eventMetas) {
        try {
          eventMeta.setRemark("recover");
          eventMetaService.modifyEventMetaStatus2W(eventMeta);

          eventQueue.put(eventMeta);
          log.info("{} have puted into queue",eventMeta);

          checkRecoverProgress.submit(createEventRecoverTask());
        } catch (Exception e) {
          log.error("recover " + eventMeta.getObjectKey() + "fail ", e);
          throw e;
        }
      }
      log.info("end try to reset ordered-event state:[P、F->W]");
    }

  }

  public ServerConfig getServerConfig() {
    return serverConfig;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }

  public BlockingQueue<EventMeta> getEventQueue() {
    return eventQueue;
  }

  public BlockingQueue<EventDataInfo> getEvenDataQueue() {
    return evenDataQueue;
  }

  public BlockingQueue<BuildJsonResult> getBuildJsonQueue() {
    return buildJsonQueue;
  }
}
